package com.jdrx.sw.sewage.task.xxjob;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.jdrx.ocp.dao.UserDAO;
import com.jdrx.platform.commons.rest.exception.BizException;
import com.jdrx.sw.sewage.beans.dto.AiModelQueryDTO;
import com.jdrx.sw.sewage.beans.entity.*;
import com.jdrx.sw.sewage.beans.vo.AiForecastWeeklyVO;
import com.jdrx.sw.sewage.beans.vo.DoAndMlssVO;
import com.jdrx.sw.sewage.common.constant.Constant;
import com.jdrx.sw.sewage.common.enums.EAiStatisticsType;
import com.jdrx.sw.sewage.common.enums.EIntelligenceConf;
import com.jdrx.sw.sewage.common.enums.InOutType;
import com.jdrx.sw.sewage.common.enums.MessageSubscribeType;
import com.jdrx.sw.sewage.common.helper.HttpClientHelper;
import com.jdrx.sw.sewage.common.util.LocalDateTimeUtils;
import com.jdrx.sw.sewage.common.util.NumberUtil;
import com.jdrx.sw.sewage.common.util.StringUtil;
import com.jdrx.sw.sewage.dao.*;
import com.jdrx.sw.sewage.service.*;
import com.jdrx.sw.sewage.service.common.CommandService;
import org.apache.commons.jexl3.JexlContext;
import org.apache.commons.jexl3.JexlEngine;
import org.apache.commons.jexl3.JexlScript;
import org.apache.commons.jexl3.MapContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

import static com.jdrx.sw.sewage.common.constant.Constant.*;
import static com.jdrx.sw.sewage.common.enums.DataAnalysisType.WD;

@Component
public class AsyncSewageJob {
    private static final Logger logger = LoggerFactory.getLogger(SewageJob.class);
    @Value("${spring.application.name}")
    private String platformCode;
    @Value("${forecast.pumpChokeUrl}")
    private String pumpChokeUrl;

    @Autowired
    private SequenceDefineDAO sequenceDefineDao;
    @Autowired
    private StationReportServiceImpl stationReportServiceImpl;
    @Autowired
    private StationReportAnalysisServiceImpl stationReportAnalysisServiceImpl;
    @Autowired
    private DeviceTimeDataDAO deviceTimeDataDAO;
    @Autowired
    private MessageSubscribeServiceImpl messageSubscribeServiceImpl;
    @Autowired
    private UserDAO userDAO;
    @Autowired
    private StationDAO stationDAO;
    @Autowired
    private AlarmDAO alarmDAO;
    @Autowired
    private DevStorageDAO devStorageDAO;
    @Autowired
    private GatherDevPointDAO gatherDevPointDAO;
    @Autowired
    private MessageSubscribeDAO messageSubscribeDAO;
    @Autowired
    private StationReportAnalysisServiceImpl reportAnalysisServiceImpl;
    @Autowired
    private CommandService commandService;
    @Autowired
    private AiModelParamDAO aiModelParamDAO;
    @Autowired
    private AiForecastDAO aiForecastDAO;
    @Autowired
    private AiReportDAO aiReportDAO;
    @Autowired
    private JexlEngine jexlEngine;
    @Autowired
    private AiModelParamServiceImpl aiModelParamServiceImpl;
    @Autowired
    private AiForecastServiceImpl aiForecastServiceImpl;
    @Autowired
    private AiModelDAO aiModelDAO;
    @Autowired
    private IntelligenceConfServiceImpl intelligenceConfServiceImpl;

    @Async("sewageJobExecutor")
    public void aiReport(AiModelPO aiModelPO, LocalDateTime ymdh, Date now) {
        logger.info("ai报表执行开始 - {}", aiModelPO);
        String parentName = aiModelPO.getParentName();
        List<AiReportPO> list = Lists.newArrayList();
        String parentCode = aiModelPO.getParentCode();
        if ("智能碳源".equals(parentName) || "智能除磷".equals(parentName) || "智能曝气".equals(parentName)) {
            //模型训练数据
//            AiReportPO MODEL_DATA = new AiReportPO();
//            MODEL_DATA.setCreateTime(ymdh);
//            MODEL_DATA.setName(Constant.MODEL_DATA);
//            MODEL_DATA.setType(EAiStatisticsType.MODEL_DATA.getKey());
//            MODEL_DATA.setParentCode(parentCode);
//            list.add(MODEL_DATA);

            //累计清洗数据
            Map<String, Double> crrData = gatherDevPointDAO.findByModePrCode(parentName);
            List<AiModelParamPO> pos = aiModelParamDAO.findParByParentName(parentName);
            if (null != pos && pos.size() > 0) {
                pos.forEach(item -> {
                    String pointCode = item.getCode();
                    if (crrData.containsKey(pointCode)) {
                        JexlContext jexlContext = new MapContext();
                        jexlContext.set(pointCode, crrData.get(pointCode));

                        JexlScript jexlScript = jexlEngine.createScript(item.getExceptionRule());
                        // 执行Jexl表达式，得到结果
                        Integer result = (Integer) jexlScript.execute(jexlContext);
                        if (1 == result) {
                            AiReportPO CLEAN_DATA = new AiReportPO();
                            CLEAN_DATA.setCreateTime(ymdh);
                            CLEAN_DATA.setName(item.getExceptionDescribe());
                            CLEAN_DATA.setType(EAiStatisticsType.CLEAN_DATA.getKey());
                            CLEAN_DATA.setParentCode(parentCode);
                            list.add(CLEAN_DATA);
                        }
                    }
                });
            }
        }

        if (null != list && list.size() > 0) aiReportDAO.batchInsert(list);
        logger.info("ai报表执行结束 - ：{}", aiModelPO.getCode());
    }

    @Async("sewageJobExecutor")
    public void aiModelExecute(AiModelPO aiModelPO, LocalDateTime ymdh, Date now) {
        logger.info("ai模型预测执行开始 - {}", aiModelPO);

        String code = aiModelPO.getCode();
        Integer retvalCount = aiModelPO.getRetvalCount();

        AiForecastPO crrAiForecastPO = new AiForecastPO();
        crrAiForecastPO.setForecastTime(ymdh);
        crrAiForecastPO.setModelCode(code);
        crrAiForecastPO.setRealtimeData(aiModelPO.getParValue());

        List<AiForecastPO> list = Lists.newArrayList();
        list.add(crrAiForecastPO);
        AiForecastPO oneAiForecastPO = null;
        AiForecastPO twoAiForecastPO = null;
        LocalDateTime ymdh1 = LocalDateTimeUtils.stringToLocalDateTime(LocalDateTimeUtils.dateCURRToString2(LocalDateTimeUtils.addHour(now, 1)));
        LocalDateTime ymdh2 = LocalDateTimeUtils.stringToLocalDateTime(LocalDateTimeUtils.dateCURRToString2(LocalDateTimeUtils.addHour(now, 2)));
        if (1 <= retvalCount) {
            oneAiForecastPO = new AiForecastPO();
            oneAiForecastPO.setForecastTime(ymdh1);
            oneAiForecastPO.setModelCode(code);
            list.add(oneAiForecastPO);
        }
        if (2 <= retvalCount) {
            twoAiForecastPO = new AiForecastPO();
            twoAiForecastPO.setForecastTime(ymdh2);
            twoAiForecastPO.setModelCode(code);
            list.add(twoAiForecastPO);
        }
        aiForecastDAO.batchInsert(list);

        List<AiModelParamPO> AiModelParamList = aiModelParamDAO.findParByModeCode(code);
        if (null == AiModelParamList && AiModelParamList.size() <= 0) {
            logger.info("ai模型预测执行开始 - 模型参数配置为空，不处理ai调用：{}", code);
            return;
        }
        JSONObject par = null;
        Map<String, List<AiModelParamPO>> map = AiModelParamList.stream().collect(Collectors.groupingByConcurrent(AiModelParamPO::getTimeRule));
        Map<String, Double> parData = Maps.newHashMap();
        Map<String, Double> realTimeData = null;
        Map<String, Double> oneHourSumData = null;
        Map<String, Double> oneHourAvgData = null;
        Map<String, Double> twoHourSumData = null;
        Map<String, Double> twoHourAvgData = null;
        Map<String, Double> threeHourAvgData = null;
        Map<String, Double> sevenHourAvgData = null;
        Map<String, Double> nineHourAvgData = null;
        Map<String, Double> sevenDayAvgData = null;
        Map<String, Double> waterQualityForecast = null;
        for (Map.Entry<String, List<AiModelParamPO>> entry : map.entrySet()) {
            String mapKey = entry.getKey();
            List<AiModelParamPO> mapValue = entry.getValue();
            if ("0H".equals(mapKey)) {
                realTimeData = gatherDevPointDAO.findByValueAndVal(mapValue);
            } else {
                Map<String, List<AiModelParamPO>> map2 = mapValue.stream().collect(Collectors.groupingByConcurrent(AiModelParamPO::getCalculationRule));
                if (map2 == null || map2.size() == 0) continue;
                for (Map.Entry<String, List<AiModelParamPO>> entry2 : map2.entrySet()) {
                    String mapKey2 = entry2.getKey();
                    AiModelQueryDTO dto = new AiModelQueryDTO();
                    dto.setAiModelParamPOs(entry2.getValue());
                    if ("avg".equals(mapKey2)) {
                        if ("1H".equals(mapKey)) {
                            dto.setTime("'1 H'");
                            dto.setTimeRule("1H");
                            oneHourAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                        if ("2H".equals(mapKey)) {
                            dto.setTime("'2 H'");
                            dto.setTime2("'1 H'");
                            dto.setTimeRule("2H");
                            twoHourAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                        if ("3H".equals(mapKey)) {
                            dto.setTime("'3 H'");
                            dto.setTime2("'2 H'");
                            dto.setTimeRule("3H");
                            threeHourAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                        if ("7H".equals(mapKey)) {
                            dto.setTime("'7 H'");
                            dto.setTime2("'6 H'");
                            dto.setTimeRule("7H");
                            sevenHourAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                        if ("9H".equals(mapKey)) {
                            dto.setTime("'9 H'");
                            dto.setTime2("'8 H'");
                            dto.setTimeRule("9H");
                            nineHourAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                        if ("7D".equals(mapKey)) {
                            dto.setTime("'7 D'");
                            dto.setTimeRule("7D");
                            sevenDayAvgData = deviceTimeDataDAO.findOneHourAvgPars(dto);
                        }
                    }
                    if ("sum".equals(mapKey2)) {
                        if ("1H".equals(mapKey)) {
                            dto.setTime("'1 H'");
                            oneHourSumData = deviceTimeDataDAO.findOneHourSumPars(dto);
                        }
                        if ("2H".equals(mapKey)) {
                            dto.setTime("'2 H'");
                            twoHourSumData = deviceTimeDataDAO.findOneHourSumPars(dto);
                        }
                    }
                    if ("predict".equals(mapKey2)) {
                        waterQualityForecast = deviceTimeDataDAO.findWaterQualityForecast(ymdh1);
                    }
                }
            }
        }

        if (null != realTimeData && !realTimeData.isEmpty())
            parData.putAll(realTimeData);
        else logger.info("ai模型执行：{} - 模型实时参数值为空：{}", code, parData);

        if (null != oneHourSumData && !oneHourSumData.isEmpty())
            parData.putAll(oneHourSumData);
        else logger.info("ai模型执行：{} - 模型一小时sum参数值为空：{}", code, oneHourSumData);

        if (null != oneHourAvgData && !oneHourAvgData.isEmpty())
            parData.putAll(oneHourAvgData);
        else logger.info("ai模型执行：{} - 模型一小时avg参数值为空：{}", code, oneHourAvgData);

        if (null != twoHourSumData && !twoHourSumData.isEmpty())
            parData.putAll(twoHourSumData);
        else logger.info("ai模型执行：{} - 模型两小时sum参数值为空：{}", code, twoHourSumData);

        if (null != twoHourAvgData && !twoHourAvgData.isEmpty())
            parData.putAll(twoHourAvgData);
        else logger.info("ai模型执行：{} - 模型两小时avg参数值为空：{}", code, twoHourAvgData);

        if (null != threeHourAvgData && !threeHourAvgData.isEmpty())
            parData.putAll(threeHourAvgData);
        else logger.info("ai模型执行：{} - 模型三小时avg参数值为空：{}", code, threeHourAvgData);

        if (null != sevenHourAvgData && !sevenHourAvgData.isEmpty())
            parData.putAll(sevenHourAvgData);
        else logger.info("ai模型执行：{} - 模型三小时avg参数值为空：{}", code, sevenHourAvgData);

        if (null != nineHourAvgData && !nineHourAvgData.isEmpty())
            parData.putAll(nineHourAvgData);
        else logger.info("ai模型执行：{} - 模型三小时avg参数值为空：{}", code, nineHourAvgData);


        if (null != sevenDayAvgData && !sevenDayAvgData.isEmpty())
            parData.putAll(sevenDayAvgData);
        else logger.info("ai模型执行：{} - 模型七天avg参数值为空：{}", code, sevenDayAvgData);

        if (null != waterQualityForecast && !waterQualityForecast.isEmpty())
            parData.putAll(waterQualityForecast);
        else logger.info("ai模型执行：{} - 模型水质预测参数值为空：{}", code, waterQualityForecast);

        if (null != parData && !parData.isEmpty()) {
            par = (JSONObject) JSONObject.toJSON(parData);
            par.put("time", LocalDateTimeUtils.dateCURRToString3(now));
        } else {
            logger.info("ai模型执行：{} - 模型参数值为空，不处理ai调用：{}", code, parData);
            return;
        }

        IntelligenceConfPO intelligenceConfPO = intelligenceConfServiceImpl.selectParameterConfigure(aiModelPO.getParentName(), EIntelligenceConf.MODEL_SET.getKey());
        if (null != intelligenceConfPO && null != intelligenceConfPO.getConfParam())
            par.putAll(intelligenceConfPO.getConfParam());
        else if (null != intelligenceConfPO && null == intelligenceConfPO.getConfParam())
            logger.info("ai模型执行：{} - 模型手动参数值为空，不处理ai调用：{}", code, parData);

        if (null == parData || parData.isEmpty()) {
            logger.info("ai模型执行：{} - 模型参数值为空，不处理ai调用：{}", code, parData);
            return;
        }

        logger.info("ai调用执行开始：{} - 模型参数和值为：{}", code, parData);
        String result = HttpClientHelper.sendPost(aiModelPO.getUrl(), JSON.toJSONString(par));
        if (null == result || StringUtils.isEmpty(result)) {
            logger.info("ai调用执行返回为空：{} - 值为：{}", code, result);
            return;
        }

        JSONArray resultArr = JSON.parseArray(result);
        if (resultArr.isEmpty()) {
            logger.info("ai调用执行返回为空：{} - 值为：{}", code, result);
            return;
        }

        Double first = null;
        Double two = null;
        Double three = null;
        Double four = null;
        Double five = null;
        list.clear();
        if (1 <= retvalCount) {
            first = resultArr.getDouble(0);
            oneAiForecastPO.setForecastData(first);
        }
        if (2 <= retvalCount) {
            two = resultArr.getDouble(1);
            twoAiForecastPO.setForecastData(two);
            list.add(twoAiForecastPO);
        }
        //目前按照除磷约定的，第三个值为策略
        if (3 <= retvalCount) {
            three = resultArr.getDouble(2);
        }
        if (4 <= retvalCount) {
            four = resultArr.getDouble(3);
            oneAiForecastPO.setForecastData2(four);
            list.add(oneAiForecastPO);
        }
        if (5 <= retvalCount) {
            five = resultArr.getDouble(4);
            oneAiForecastPO.setElectricQuantity(five);
            list.add(oneAiForecastPO);
        }
        aiForecastDAO.batchUpdate(list);

        //ai对话框
        List<AiReportPO> aiList = Lists.newArrayList();
        aiReportEx(aiModelPO, ymdh, ymdh1, ymdh2, first, aiList);

        if (null != aiList && aiList.size() > 0) aiReportDAO.batchInsert(aiList);
        logger.info("ai模型预测执行结束 - ：{}", code);
    }

    @Async("sewageJobExecutor")
    public void deviceTimeDataBackup2(String dataNode, LocalDateTime now) {
        logger.info(dataNode + "实时数据备份 - 开始，执行时间{}", LocalDateTime.now());
        Boolean result = deviceTimeDataDAO.deviceTimeDataBackup2();
        logger.info(dataNode + "实时数据备份 - 完成，执行时间{}，执行状态{}", LocalDateTime.now(), result);
    }

    @Async("sewageJobExecutor")
    public void stationWaterQuality2(String dataNode) {
        Date now = new Date();
        logger.info(dataNode + "数据大屏水质数据拉取 - 开始，执行时间{}", LocalDateTime.now());
        List<String> iotGatewayNums = gatherDevPointDAO.findWaterQualityPoi(dataNode, null);
        if (iotGatewayNums.size() > 0) {
            iotGatewayNums.forEach(item -> {
                List<String> points = gatherDevPointDAO.findWaterQualityPoi(dataNode, item);
                commandService.CommandLssuePullData(item, now, points, dataNode);
            });
        }
        logger.info(dataNode + "数据大屏水质数据拉取 - 执行完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void alarmNumInitValueJob2(String dataNode) {
        logger.info(dataNode + "初始化告警编号初始值为1开始，执行时间{}", LocalDateTime.now());
        SequenceDefinePO sequenceDefinePO = sequenceDefineDao.findByKey(Constant.ALARM_NUM_SN_KEY, platformCode, dataNode);
        if (Objects.nonNull(sequenceDefinePO)) {
            sequenceDefineDao.update(sequenceDefinePO.getId(),
                    1L, new Date(), sequenceDefinePO.getRemarks(), dataNode);
        }
        logger.info(dataNode + "初始化告警编号初始值为1 - 执行完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void deviceTimeDataHourFilterJob2(Integer reportType, LocalDateTime now2, String dataNode, String strTime, Date startTime) throws BizException {
        logger.info("清晰数据分析-日报表- 每天整点 - 执行开始，执行时间{}", LocalDateTime.now());
        Map<String, Object> params = Maps.newHashMap();
        Date endTime = LocalDateTimeUtils.reduceHour(0);
        params.put("startTime", startTime);
        params.put("endTime", endTime);
        params.put("dataNode", dataNode);
        params.put("strTime", strTime);
        List<DeviceTimeDataPO> dataPOS = stationReportServiceImpl.selectRealTimeLastPacksByParams(params);
        Map<Long, List<DeviceTimeDataPO>> groupBy = dataPOS.stream().collect(Collectors.groupingBy(DeviceTimeDataPO::getStationId));

        List<StationReportAnalysisPO> dataList1 = new ArrayList<>();
        for (Long mapKey : groupBy.keySet()) {
            //日报表
            StationReportAnalysisPO JSLJLLPO = new StationReportAnalysisPO();
            JSLJLLPO.setStationId(mapKey);
            JSLJLLPO.setCreateTime(now2);
            JSLJLLPO.setInOut(InOutType.IN.getKey());
            JSLJLLPO.setReportType(reportType);

            StationReportAnalysisPO CSLJLLPO = new StationReportAnalysisPO();
            CSLJLLPO.setStationId(mapKey);
            CSLJLLPO.setCreateTime(now2);
            CSLJLLPO.setInOut(InOutType.OUT.getKey());
            CSLJLLPO.setReportType(reportType);

            StationReportAnalysisPO DNBPO = new StationReportAnalysisPO();
            DNBPO.setStationId(mapKey);
            DNBPO.setCreateTime(now2);
            DNBPO.setInOut(InOutType.DNB.getKey());
            DNBPO.setReportType(reportType);

            // 碳源加药
            StationReportAnalysisPO TYJYLJLLPO = new StationReportAnalysisPO();
            TYJYLJLLPO.setStationId(mapKey);
            TYJYLJLLPO.setCreateTime(now2);
            TYJYLJLLPO.setInOut(InOutType.TYJY.getKey());
            TYJYLJLLPO.setReportType(reportType);

            // PAC加药
            StationReportAnalysisPO PACJYLJLLPO = new StationReportAnalysisPO();
            PACJYLJLLPO.setStationId(mapKey);
            PACJYLJLLPO.setCreateTime(now2);
            PACJYLJLLPO.setInOut(InOutType.PACJY.getKey());
            PACJYLJLLPO.setReportType(reportType);

            List<DeviceTimeDataPO> deviceTimeDataPOS = groupBy.get(mapKey);
            Double JSLJLL_R = null;
            Double CSLJLL_R = null;
            Double DNB_R = null;
            Double TYJYSSLL = null;
            Double PACJYLJLL = null;
            if (null != deviceTimeDataPOS && deviceTimeDataPOS.size() > 0) {
                for (int j = 0; j < deviceTimeDataPOS.size(); j++) {
                    DeviceTimeDataPO item = deviceTimeDataPOS.get(j);
                    JSONObject jsonVal = item.getJsonVal();
                    for (String jsonKey : jsonVal.keySet()) {
                        //日报表
                        String[] keyArr_R = jsonKey.split("_");
                        String devStr = StringUtil.removeStrMatcher(keyArr_R[1]);
                        Double value = jsonVal.getDouble(jsonKey);

                        if (devStr.equals(Constant.DNB)) {
                            Map<String, Object> params2 = Maps.newHashMap();
                            params2.put("startTime", startTime);
                            params2.put("endTime", endTime);
                            params2.put("dataNode", dataNode);
                            params2.put("devId", item.getDevId());
                            List<DeviceTimeDataPO> deviceTimeLJLLDataPO = deviceTimeDataDAO.findRealTimeFirstPacksByParams(params2);
                            Double fValue = deviceTimeLJLLDataPO.get(0).getJsonVal().getDouble(jsonKey);
                            if (null != DNB_R) DNB_R = new BigDecimal(DNB_R).add(sub(value, fValue)).doubleValue();
                            else DNB_R = sub(value, fValue).doubleValue();
                        }

                        String[] keyArr = jsonKey.split("_");
                        String dev = keyArr_R[2];

                        String pusStr = StringUtil.removeStrMatcher(keyArr[0]);
                        value = new BigDecimal(NumberUtil.precision(new BigDecimal(value), 2)).doubleValue();
                        if (pusStr.equals(CSQ)) {
                            if (dev.contains(COD)) CSLJLLPO.setCod(value);
                            if (dev.contains(TN)) CSLJLLPO.setTn(value);
                            if (dev.contains(TP)) CSLJLLPO.setTp(value);
                            if (dev.contains(NH3N)) CSLJLLPO.setNh3n(value);
                            if (StringUtil.removeStrMatcher(dev).equals(SS)) CSLJLLPO.setSs(value);
                            if (dev.contains(WD.getValue())) CSLJLLPO.setTemperature(value);
                            if (dev.contains(PH)) CSLJLLPO.setPh(value);
                            if (dev.contains(LJLL)) {
                                Map<String, Object> params2 = Maps.newHashMap();
                                params2.put("startTime", startTime);
                                params2.put("endTime", endTime);
                                params2.put("dataNode", dataNode);
                                params2.put("devId", item.getDevId());
                                List<DeviceTimeDataPO> deviceTimeLJLLDataPO = deviceTimeDataDAO.findRealTimeFirstPacksByParams(params2);
                                Double fValue = deviceTimeLJLLDataPO.get(0).getJsonVal().getDouble(jsonKey);
                                if (null != CSLJLL_R)
                                    CSLJLL_R = new BigDecimal(CSLJLL_R).add(sub(value, fValue)).doubleValue();
                                else CSLJLL_R = sub(value, fValue).doubleValue();
                            }
                        }
                        if (pusStr.equals(JSK)) {
                            if (dev.contains(COD)) JSLJLLPO.setCod(value);
                            if (dev.contains(TN)) JSLJLLPO.setTn(value);
                            if (dev.contains(TP)) JSLJLLPO.setTp(value);
                            if (dev.contains(NH3N)) JSLJLLPO.setNh3n(value);
                            if (StringUtil.removeStrMatcher(dev).equals(SS)) JSLJLLPO.setSs(value);
                            if (dev.contains(WD.getValue())) JSLJLLPO.setTemperature(value);
                            if (dev.contains(PH)) JSLJLLPO.setPh(value);
                            if (dev.contains(LJLL)) {
                                Map<String, Object> params2 = Maps.newHashMap();
                                params2.put("startTime", startTime);
                                params2.put("endTime", endTime);
                                params2.put("dataNode", dataNode);
                                params2.put("devId", item.getDevId());
                                List<DeviceTimeDataPO> deviceTimeLJLLDataPO = deviceTimeDataDAO.findRealTimeFirstPacksByParams(params2);
                                Double fValue = deviceTimeLJLLDataPO.get(0).getJsonVal().getDouble(jsonKey);
                                /*double v = JSLJLL_R.get() + (value - fValue);
                                JSLJLL_R.set(v);*/
                                if (null != JSLJLL_R)
                                    JSLJLL_R = new BigDecimal(JSLJLL_R).add(sub(value, fValue)).doubleValue();
                                else JSLJLL_R = sub(value, fValue).doubleValue();
                            }
                        }
                        //碳源药耗
                        if (pusStr.contains(TYJY)) {
                            if (dev.contains(SSLL)) {
                                if (null != TYJYSSLL)
                                    TYJYSSLL = new BigDecimal(TYJYSSLL).add(new BigDecimal(value)).doubleValue();
                                else TYJYSSLL = new BigDecimal(value).doubleValue();
//                                TYJYSSLL.set(TYJYSSLL.get() + value);
                            }
                        }
                        //碳源药耗
                        if (pusStr.contains(PACJY)) {
                            if (dev.contains(SSLL)) {
                                if (null != PACJYLJLL)
                                    PACJYLJLL = new BigDecimal(PACJYLJLL).add(new BigDecimal(value)).doubleValue();
                                else PACJYLJLL = new BigDecimal(value).doubleValue();
//                                PACJYLJLL.set(PACJYLJLL.get() + value);
                            }
                        }
                    }
                }
            }
            JSLJLLPO.setLjll(null == JSLJLL_R ? null : new BigDecimal(NumberUtil.precision(new BigDecimal(JSLJLL_R), 2)).doubleValue());
            CSLJLLPO.setLjll(null == CSLJLL_R ? null : new BigDecimal(NumberUtil.precision(new BigDecimal(CSLJLL_R), 2)).doubleValue());
            DNBPO.setDnb(null == DNB_R ? null : new BigDecimal(NumberUtil.precision(new BigDecimal(DNB_R), 2)).doubleValue());
            TYJYLJLLPO.setLjll(null == TYJYSSLL ? null : new BigDecimal(NumberUtil.precision(new BigDecimal(TYJYSSLL), 2)).doubleValue());
            PACJYLJLLPO.setLjll(null == PACJYLJLL ? null : new BigDecimal(NumberUtil.precision(new BigDecimal(PACJYLJLL), 2)).doubleValue());

            dataList1.add(JSLJLLPO);
            dataList1.add(CSLJLLPO);
            dataList1.add(DNBPO);
            dataList1.add(TYJYLJLLPO);
            dataList1.add(PACJYLJLLPO);
        }
        stationReportAnalysisServiceImpl.batchInsertHistory(dataList1, dataNode);
        logger.info("清晰数据分析-日报表- 每天整点 - 执行完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void operateBriefPush2(String dataNode) {
        logger.info(dataNode + "运营简报推送 开始，执行时间{}", LocalDateTime.now());
        String[] data = LocalDateTimeUtils.getYesterday().split("-");

        List<StationPO> stationPOS = stationDAO.selectStationList(dataNode);
        stationPOS.forEach(station -> {
            Integer alarmNum = alarmDAO.countLastDayAlarmNum(station.getId(), dataNode);
            String operationAllAnalysis = reportAnalysisServiceImpl.operationAllAnalysis(station.getId(), dataNode);

            //push message
            try {
                List<Long> userIds = messageSubscribeDAO.selectMessageByStationIdAndType(station.getId(), MessageSubscribeType.BRIEF_REPORT_PUSH.getKey(), dataNode);
                if (null != userIds && userIds.size() > 0) {
                    userDAO.findOpenId(userIds, dataNode).forEach(item -> {
                        try {
                            if (StringUtils.isNotEmpty(item))
                                messageSubscribeServiceImpl.pushAlarm(item,
                                        null,
                                        data[0] + "月" + data[1] + "日的报告已生成，水质达标" + operationAllAnalysis + "%，当日出现" + alarmNum + "告警",
                                        "运营日报通知",
                                        "简报已生成，点击查看详情",
                                        MessageSubscribeType.BRIEF_REPORT_PUSH.getKey());

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    });
                }
            } catch (Exception e) {
                logger.error("运营简报推送失败！", e);
            }
        });
        logger.info(dataNode + "运营简报推送 - 执行完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void maintenanceCycleDay2(String dataNode) {
        logger.info(dataNode + "维保周期天更新 开始，执行时间{}", LocalDateTime.now());
        List<DevStoragePO> devStoragePOs = devStorageDAO.findAllList(dataNode);
        devStoragePOs.forEach(item -> {
            devStorageDAO.maintenanceCycleDay(dataNode, item.getId(), item.getCycleTime() - 1);
        });
        logger.info(dataNode + "维保周期天更新 完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void maintenanceCycleHour2(String dataNode) {
        logger.info(dataNode + "维保周期小时更新 开始，执行时间{}", LocalDateTime.now());
        List<DevStoragePO> devStoragePOs = devStorageDAO.findCollectAllList(dataNode);
        devStoragePOs.forEach(item -> {
            devStorageDAO.maintenanceCycleHour(dataNode, item.getId(), item.getWorkTime() - 1);
        });
        logger.info(dataNode + "维保周期小时更新 完成，执行时间{}", LocalDateTime.now());
    }

    @Async("sewageJobExecutor")
    public void aiModeWeekReportJob(String dataNode) {
        logger.info("ai周报生成执行开始");
        List<AiModelPO> aiModelPOList = aiModelDAO.findAiModelByCron(null);
        if (null == aiModelPOList && aiModelPOList.size() <= 0) {
            logger.info("ai模型执行 - 模型配置错误或为空，不处理ai调用{}", LocalDateTime.now());
            return;
        }
        LocalDateTime ymdh = LocalDateTime.now();
        List<AiReportPO> list = Lists.newArrayList();
        aiModelPOList.forEach(item -> {
            String parentName = item.getParentName();
            String identifier = item.getIdentifier();
            if ("智能碳源".equals(parentName) || "智能除磷".equals(parentName) || "智能曝气".equals(parentName)) {
                //智能碳源分析报告
                AiReportPO MODEL_DATA2 = new AiReportPO();
                MODEL_DATA2.setCreateTime(ymdh);
                MODEL_DATA2.setName(Constant.MODEL_ANALYSIS_DATA);
                MODEL_DATA2.setType(EAiStatisticsType.MODEL_DATA.getKey());
                MODEL_DATA2.setParentCode(item.getParentCode());

                AiReportPO ANALYSIS_REPORT_DATA = new AiReportPO();
                ANALYSIS_REPORT_DATA.setCreateTime(ymdh);
                ANALYSIS_REPORT_DATA.setName(parentName + ANALYSIS_DATA + LocalDateTimeUtils.localDateTime3String(ymdh));
                ANALYSIS_REPORT_DATA.setType(EAiStatisticsType.ANALYSIS_REPORT_DATA.getKey());
                ANALYSIS_REPORT_DATA.setParentCode(item.getParentCode());
                Double v = aiReportDAO.findLastWeek(item.getParentName());
                ANALYSIS_REPORT_DATA.setReportAbstract(String.format("节省药剂%s", v == null ? "-" : v * 100) + "%");

                Integer countNum = aiReportDAO.countNum(EAiStatisticsType.ANALYSIS_REPORT_DATA.getKey(), parentName) + 1;
                int num = LengthNum(countNum);
                StringBuilder number = new StringBuilder(identifier);
                if (1 == num) number.append("000");
                if (2 == num) number.append("00");
                if (3 == num) number.append("0");
                number.append(countNum);
                ANALYSIS_REPORT_DATA.setNumber(number.toString());

                JSONObject json = new JSONObject();
                JSONObject headJson = new JSONObject();
                //只适合一体机一个站点（产品设计无多站点情况！！！）
                StationPO stationPO = stationDAO.selectStation();
                LocalDateTime crTime = LocalDateTimeUtils.stringToLocalDateTime(LocalDateTimeUtils.dateCURRToString4(new Date()));
                String startTime = LocalDateTimeUtils.localDate2String(LocalDateTimeUtils.stepOrAddDay(crTime, -7));
                String endTime = LocalDateTimeUtils.localDate2String(LocalDateTimeUtils.stepOrAddDay(crTime, -1));
                headJson.put("forecastCycle", startTime + " 至 " + endTime);
                headJson.put("stationName", stationPO.getName());
                headJson.put("reportTime", LocalDateTimeUtils.localDateTime2String(ymdh));
                headJson.put("stationAddress", stationPO.getAddr());
                headJson.put("reportNum", number.toString());

                if ("智能曝气".equals(parentName)) {
                    Double electricQuantity = aiForecastServiceImpl.findElectricQuantity(startTime, endTime);
                    Double cNeutralize = null == electricQuantity ? null : (electricQuantity * 0.997);
                    headJson.put("cNeutralize", cNeutralize);
                    ANALYSIS_REPORT_DATA.setCNeutralize(cNeutralize);

                    List<DoAndMlssVO> dos = deviceTimeDataDAO.findDOAndMLSS(startTime, endTime);
                    if (null != dos && dos.size() > 0) {
                        List newList = new ArrayList();
                        Map<String, List<DoAndMlssVO>> collect = dos.stream().collect(Collectors.groupingByConcurrent(DoAndMlssVO::getPointV));
                        for (Map.Entry<String, List<DoAndMlssVO>> ent : collect.entrySet()) {
                            String key = ent.getKey();
                            List<DoAndMlssVO> value = ent.getValue();
                            if (key.contains("_MLSS_")) json.put("mlssData", value);
                            if (key.contains("_DO_")) newList.add(value);
                        }
                        json.put("doData", newList);
                    } else {
                        json.put("doData", null);
                        json.put("mlssData", null);
                    }
                }
                json.put("headData", headJson);
                json.put("statisticsData", aiModelParamServiceImpl.findStatisticsNum(parentName));
                json.put("reportData", aiForecastServiceImpl.findWeekCarbonSourceForecast(crTime, parentName));
                AiForecastWeeklyVO conclusionData = aiForecastServiceImpl.findConclusion(parentName);
                json.put("conclusionData", null == conclusionData ? new AiForecastWeeklyVO() : conclusionData);

                ANALYSIS_REPORT_DATA.setReportData(json);

                list.add(MODEL_DATA2);
                list.add(ANALYSIS_REPORT_DATA);
            }
        });
        if (null != list && list.size() > 0) aiReportDAO.batchInsert(list);
        logger.info("ai周报生成执行结束");
    }

    private void aiReportEx(AiModelPO aiModelPO, LocalDateTime ymdh, LocalDateTime ymdh1, LocalDateTime ymdh2, Double first, List<AiReportPO> aiList) {
        String parentName = aiModelPO.getParentName();
        if ("智能碳源".equals(parentName) || "智能除磷".equals(parentName) || "智能曝气".equals(parentName)) {
            //ai对话框
            //当一个时间周期内结束，实际投加的药剂量没有达到预测的药剂量，给出提示：
            //格式：“1#泵实际投加药剂量XXX，比预测值±XXX量；
            AiReportPO AI_WINDOW_DATA = new AiReportPO();
            AI_WINDOW_DATA.setCreateTime(ymdh);
            AI_WINDOW_DATA.setName(String.format(AIL_YC_JY_DATA, aiModelPO.getName(), ymdh1.getHour(), ymdh2.getHour(), Double.valueOf(NumberUtil.precision(new BigDecimal(first), 3))));
            AI_WINDOW_DATA.setType(EAiStatisticsType.AI_WINDOW_DATA.getKey());
            AI_WINDOW_DATA.setParentCode(aiModelPO.getParentCode());
            AI_WINDOW_DATA.setColour("red");
            //aiReportDAO.insert(AI_WINDOW_DATA);
            aiList.add(AI_WINDOW_DATA);

            //ai对话框
            //当一个时间周期内结束，实际投加的药剂量没有达到预测的药剂量，给出提示：
            //格式：“1#泵实际投加药剂量XXX，比预测值±XXX量；
            AiForecastPO aiForecastPO = aiForecastDAO.fingByCrr(ymdh, aiModelPO.getCode());
            Double realtimeData = aiForecastPO.getRealtimeData();
            Double forecastData = aiForecastPO.getForecastData();
            if (null != realtimeData && null != forecastData) {
                Double differenceVal = Double.valueOf(NumberUtil.precision(new BigDecimal(realtimeData - forecastData), 2));
                AiReportPO AI_WINDOW_DATA2 = new AiReportPO();
                AI_WINDOW_DATA2.setCreateTime(ymdh);
                AI_WINDOW_DATA2.setName(String.format(AIL_CRR_JY_DATA, aiModelPO.getName(),
                        Double.valueOf(NumberUtil.precision(new BigDecimal(realtimeData), 2)),
                        differenceVal >= 0 ? ("+" + differenceVal) : differenceVal));
                AI_WINDOW_DATA2.setType(EAiStatisticsType.AI_WINDOW_DATA.getKey());
                AI_WINDOW_DATA2.setParentCode(aiModelPO.getParentCode());
                AI_WINDOW_DATA2.setColour("red");
                //aiReportDAO.insert(AI_WINDOW_DATA2);
                aiList.add(AI_WINDOW_DATA2);
            }
        }
    }

    private BigDecimal sub(Double value, Double fValue) {
        return new BigDecimal(value).subtract(new BigDecimal(fValue));
    }

    public int LengthNum(int num) {
        int count = 0; //计数
        while (num >= 1) {
            num /= 10;
            count++;
        }
        return count;
    }
}
